package com.my.controller;

import com.my.domain.User;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

@RestController
@RequestMapping("/demo")
public class SendController {

    //springboot已经创建好RocketMQTemplate，调用时会自动建立/断开MQ连接
    @Autowired
    RocketMQTemplate rocketMQTemplate;

    @GetMapping("/sendStr")
    public String send(){//发送字符串
        //发送逻辑
        String msg = "hello world zlz";
        //不能用原生Message类
        //Message message = new Message("topic1","tag1", msg.getBytes());
        rocketMQTemplate.convertAndSend("topic10", msg);//convertAndSend会在底层转换为字节数组

        return "success";
    }

    @GetMapping("/sendUser")
    public String send2(){//发送对象
        //发送逻辑
        User user =new User("张三",18);
        //不能用原生Message类
        //Message message = new Message("topic1","tag1", msg.getBytes());
        rocketMQTemplate.convertAndSend("topic10", user);//convertAndSend会在底层转换为字节数组
        return "success";
    }

    @GetMapping("/syncSend")
    public String send3(){//发送同步消息
        //发送逻辑
        User user =new User("张三",18);
        //不能用原生Message类
        //Message message = new Message("topic1","tag1", msg.getBytes());
        SendResult sendResult = rocketMQTemplate.syncSend("topic10", user);//convertAndSend会在底层转换为字节数组
        return sendResult.toString();
    }
    @GetMapping("/asyncSend")
    public String send4(){//发送异步消息
        //发送逻辑
        User user =new User("张三",18);
        //不能用原生Message类
        //Message message = new Message("topic1","tag1", msg.getBytes());
        rocketMQTemplate.asyncSend("topic10", user, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送异步消息成功："+sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送异步消息失败");
                throwable.printStackTrace();
            }
        },1000);
        return "发送完成，成功与否未知";
    }

    @GetMapping("/sendOneWay")
    public String send5(){//发送单向消息
        //发送逻辑
        User user =new User("张三",18);
        //不能用原生Message类
        //Message message = new Message("topic1","tag1", msg.getBytes());
        rocketMQTemplate.sendOneWay("topic10", user);
        return "发送完成，成功与否未知";
    }

    @GetMapping("/sendDelay")
    public String send6(){//发送延时消息
        //发送逻辑
        User user =new User("张三",18);
        //不能用原生Message类
        //Message message = new Message("topic1","tag1", msg.getBytes());
        SendResult sendResult = rocketMQTemplate.syncSend("topic10", MessageBuilder.withPayload(user).build(), 2000, 3);
        return sendResult.toString();
    }

    @GetMapping("/senBatchUser")
    public String send7(){//发送批量消息
        //发送逻辑
        User user =new User("张三",18);
        //不能用原生Message类
        //Message message = new Message("topic1","tag1", msg.getBytes());
        List<Message> msgList = new ArrayList<>();
        msgList.add(new Message("topic10","tag1", "abc".getBytes()));
        msgList.add(new Message("topic10","tag1", "def".getBytes()));
        msgList.add(new Message("topic10","tag1", "ghi".getBytes()));
        SendResult sendResult = rocketMQTemplate.syncSend("topic10", msgList, 1000);
        return sendResult.toString();
    }
}
